#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "msg_queue_peer.h"
#include "shmem.h"
#include <sqlite3.h>
#include <time.h>
#include <stdbool.h>
void mqtt_send(char *p);

#define CLIENTID "ExampleClientPub"
#define TOPIC_SUBSCRIBE "/app/control/down"
#define TOPIC_PUBLISH "/app/data/up"
#define QOS 1
#define TIMEOUT 10000L

cJSON *allList;
sqlite3 *db; // 数据库
char *errmsg;

volatile MQTTClient_deliveryToken deliveredtoken;

MQTTClient_deliveryToken token;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient client;
char ADDRESS[32]; // ip、端口号
int PERIOD = 0;   // 周期

union val_t
{
    int b_val;   // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
};

struct std_node
{
    int key;             // 唯一键值
    int type;            // 数据点类型
    int dev_type;        // 数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; // 变化上报后需要更新旧值
    union val_t new_val; // 从共享内存取出最新数据，放到new_val中
    int ret;             // 默认为-1，采集成功后设置为0，采集失败再置-1
};

// 消息队列格式
struct msg
{
    long type; // 区分modbus  stm32
    int key;
    union val_t val;
};

// int *num = NULL;
int *total = NULL;
static struct std_node *shm_data; // 存储共享内存数据

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    deliveredtoken = dt;
}

// 发送所有数据，json格式
void sendAllData()
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "type", 1);
    cJSON_AddNumberToObject(root, "result", 0);

    cJSON *arr = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", arr);

    char buf[32];
    if (total != NULL)
        for (int i = 0; i < *total; i++)
        {
            cJSON *data1 = cJSON_CreateObject();
            cJSON_AddItemToArray(arr, data1);

            if (shm_data[i].type == 1)
                sprintf(buf, "%d", shm_data[i].new_val.i_val );
            else if (shm_data[i].type == 2)
                sprintf(buf, "%d", shm_data[i].new_val.i_val);
            else
                sprintf(buf, "%.2f", shm_data[i].new_val.f_val);

            cJSON_AddNumberToObject(data1, "key", shm_data[i].key);
            cJSON_AddStringToObject(data1, "val", buf);
            memset(buf, 0, sizeof(buf));
        }
    char *p = cJSON_Print(root);
    // 发送
    mqtt_send(p);
    printf("全部数据上报完成\n");
    free(p);
    cJSON_Delete(root);
}

// 命令发送给消息队列
void setcontrol(cJSON *key, cJSON *val)
{
    struct msg set_msg;
    set_msg.key = key->valueint;
    for (int i = 0; i < *total; i++)
    {
        if (shm_data[i].key == key->valueint)
        {
            if (shm_data[i].dev_type == 111)
            {
                set_msg.type = 111;
            }
            else if (shm_data[i].dev_type == 222)
            {
                set_msg.type = 222;
            }

            set_msg.val.b_val =atoi(val->valuestring);
            printf("key=%d\n", set_msg.key);
            printf("type=%ld\n", set_msg.type);
            printf("val=%s\n", set_msg.val.b_val ? "true" : "false");
            // 发送
            msg_queue_send("msg", &set_msg, sizeof(set_msg), 0);
        }
    }
    printf("setcontrol okk\n");
}

// 修改上报模式
void change_mode(int mode, int period)
{
    cJSON *REPORT = cJSON_GetObjectItem(allList, "report");
    cJSON_ReplaceItemInObject(REPORT, "type", cJSON_CreateNumber(mode));
    if (mode == 2)
    {
        cJSON_ReplaceItemInObject(REPORT, "period", cJSON_CreateNumber(period));
    }
    char *p = cJSON_Print(allList);
    FILE *fp;
    fp = fopen("./node.json", "w");
    if (fp == NULL)
    {
        perror("fopen err");
        return;
    }
    fwrite(p, strlen(p), 1, fp);
    printf("change mode okk\n");
    fclose(fp);
}

// 查询某key值的历史记录
int key_his = 0;
pthread_t tid_his;
void *getHistory(void *arg)
{
    char sql[64];
    sprintf(sql, "SELECT * FROM history WHERE key=%d;", key_his);
    int KEY = key_his;
    char **result;
    int row, cloum;
    int i, j;
    if (sqlite3_get_table(db, sql, &result, &row, &cloum, &errmsg) != SQLITE_OK)
    {
        printf("err:%s\n", errmsg);
        return NULL;
    }

    char **p;
    p = result + cloum;
    char data[128];

    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "type", 4);
    cJSON_AddNumberToObject(root, "result", 0);
    cJSON_AddNumberToObject(root, "key", KEY);

    cJSON *arr = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", arr);

    for (i = row - 10; i < row; i++)
    {
        cJSON *data = cJSON_CreateObject();
        cJSON_AddItemToArray(arr, data);
        cJSON_AddStringToObject(data, "time", p[(i * cloum) + 2]);
        cJSON_AddStringToObject(data, "val", p[(i * cloum) + 1]);
    }
    char *q = cJSON_Print(root);
    mqtt_send(q);
    free(q);
    printf("历史记录:%d条\n", row);
}

// 接收
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    // 判断是不是json格式
    int is_json = 0;

    if (message->payloadlen >= 2 && ((char *)message->payload)[0] == '{' && ((char *)message->payload)[message->payloadlen - 1] == '}')
    {
        is_json = 1;
    }

    if (is_json)
    {
        // json反序列化
        cJSON *root = cJSON_Parse(message->payload);
        if (NULL == root)
        {
            perror("parse err :");
            is_json = 0;
            cJSON_Delete(root);
        }
        else
        {
            cJSON *type = cJSON_GetObjectItem(root, "type");
            // 判断类型
            if (type->valueint == 1)
            {
                // 发送全部数据
                sendAllData();
            }
            else if (type->valueint == 2)
            {
                cJSON *data = cJSON_GetObjectItem(root, "data");
                cJSON *key = cJSON_GetObjectItem(data, "key");
                cJSON *val = cJSON_GetObjectItem(data, "val");
                // 命令发送到消息队列
                setcontrol(key, val);
            }
            else if (type->valueint == 3)
            {
                cJSON *data = cJSON_GetObjectItem(root, "data");
                cJSON *type_mode = cJSON_GetObjectItem(data, "type");
                cJSON *type_period = cJSON_GetObjectItem(data, "period");
                // 修改模式
                change_mode(type_mode->valueint, type_period->valueint);
            }
            else if (type->valueint == 4)
            {
                cJSON *data = cJSON_GetObjectItem(root, "data");
                cJSON *key = cJSON_GetObjectItem(data, "key");
                key_his = key->valueint;
                pthread_create(&tid_his, NULL, getHistory, NULL);
            }
            cJSON_Delete(root);
        }
    }
    if (!is_json)
    {
        payloadptr = message->payload;
        printf("*****parse err :");
        for (i = 0; i < message->payloadlen; i++)
        {
            putchar(*payloadptr++);
        }
    }
    putchar('\n');
    // 释放消息相关内存
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

// MQTT发布
void mqtt_send(char *p)
{
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    MQTTClient_publishMessage(client, TOPIC_PUBLISH, &pubmsg, &token);
}

// 周期上报
void *up_period(void *arg)
{
    while (1)
    {
        sleep(PERIOD);
        sendAllData();
    }
}

// 变化上报
int change_flag = 0;
void *up_change(void *arg)
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "type", 1);
    cJSON_AddNumberToObject(root, "result", 0);

    cJSON *arr = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", arr);
    char buf[32];
    while (1)
    {
        for (int i = 0; i < *total; i++)
        {
            if (memcmp(&shm_data[i].old_val, &shm_data[i].new_val, sizeof(union val_t)) != 0)
            {
                change_flag = 1;
                cJSON *data1 = cJSON_CreateObject();
                cJSON_AddItemToArray(arr, data1);
                if (shm_data[i].type == 1)
                {
                    sprintf(buf, "%s", shm_data[i].new_val.i_val ? "true" : "false");
                }
                if (shm_data[i].type == 2)
                {
                    sprintf(buf, "%d", shm_data[i].new_val.i_val);
                    printf("new:%d\n", shm_data[i].new_val.i_val);
                    printf("old:%d\n", shm_data[i].old_val.i_val);
                }
                else
                {
                    sprintf(buf, "%.2f", shm_data[i].new_val.f_val);
                    printf("new:%f\n", shm_data[i].new_val.f_val);
                    printf("old:%f\n", shm_data[i].old_val.f_val);
                }
                shm_data[i].old_val = shm_data[i].new_val;
                cJSON_AddNumberToObject(data1, "key", shm_data[i].key);
                cJSON_AddStringToObject(data1, "val", buf);
                memset(buf, 0, sizeof(buf));
            }
        }
        if (change_flag)
        {
            char *p = cJSON_Print(root);
            mqtt_send(p);
            printf("数据有变化,已上报\n");
            change_flag = 0;
            free(p);
        }
    }
    cJSON_Delete(root);
}

int main(int argc, char *argv[])
{
    // 打开数据库
    if (sqlite3_open("./data.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open err:%s", sqlite3_errmsg(db));
        return -1;
    }
    printf("打开数据库成功\n");

    // 创建表
    if (sqlite3_exec(db, "create table history(key INTEGER,val TEXT,time TEXT);", NULL, NULL, &errmsg))
    {
        fprintf(stderr, " user :%s\n", errmsg);
    }

    // 初始化共享内存
    struct shm_param para;
    if (shm_init(&para, "shm", 1024) < 0)
    {
        perror("shmem init err");
        return -1;
    }
    void *node_p = shm_getaddr(&para);
    if (node_p == NULL)
    {
        perror("getaddr err");
        return 0;
    }

    // 打开并读取配置文件
    FILE *fp;
    char buf[3000];
    fp = fopen("./node.json", "r");
    if (fp == NULL)
    {
        perror("fopen node err");
        return -1;
    }
    fread(buf, 3000, 1, fp);
    fclose(fp);
    // 解析node.json文件
    // stm32
    allList = cJSON_Parse(buf);
    cJSON *STM32 = cJSON_GetObjectItem(allList, "stm32");
    cJSON *Data_32 = cJSON_GetObjectItem(STM32, "data");

    // modbus
    cJSON *Modbus = cJSON_GetObjectItem(allList, "modbus");
    cJSON *Data_modbus = cJSON_GetObjectItem(Modbus, "data");

    // 共享内存内结构体个数
    //*total = cJSON_GetArraySize(Data_32) + cJSON_GetArraySize(Data_modbus);
    total = (int *)node_p;

    shm_data = (struct std_node *)(node_p + sizeof(int));
    printf("num = %d\n", *total);

    for (int i = 0; i < cJSON_GetArraySize(Data_32); i++)
    {
        cJSON *item = cJSON_GetArrayItem(Data_32, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        shm_data[i].key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        shm_data[i].type = type->valueint;
        shm_data[i].dev_type = 111;
        shm_data[i].ret = -1;
    }
    for (int i = 0; i < cJSON_GetArraySize(Data_modbus); i++)
    {
        cJSON *item = cJSON_GetArrayItem(Data_modbus, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        shm_data[i + cJSON_GetArraySize(Data_32)].key = key->valueint;
        cJSON *type = cJSON_GetObjectItem(item, "type");
        shm_data[i + cJSON_GetArraySize(Data_32)].type = key->valueint;
        shm_data[i + cJSON_GetArraySize(Data_32)].dev_type = 2;
        shm_data[i + cJSON_GetArraySize(Data_32)].ret = -1;
    }

    // MQTT连接
    // 解析node.json 获取ip、端口号
    cJSON *mqtt_server = cJSON_GetObjectItem(allList, "mqtt_server");
    cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
    cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
    sprintf(ADDRESS, "tcp://%s:%d", addr->valuestring, port->valueint);
    // 连接
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    // 创建客户端，并且指定客户端连接的mqtt服务器地址和客户端ID
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    // 初始化连接参数
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    // 设置回调接口，只需要关注msgarrvd：消息到达后，会自动调用这个接口
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    // 连接到broker
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    // 订阅某个主题，指定订阅主题的名字，可以指定qos服务质量
    MQTTClient_subscribe(client, TOPIC_SUBSCRIBE, QOS);

    // 上报模式选择
    pthread_t tid;
    cJSON *report = cJSON_GetObjectItem(allList, "report");
    cJSON *up_mode = cJSON_GetObjectItem(report, "type");
    if (up_mode->valueint == 0)
    {
        printf("当前上报模式：不上报\n");
    }
    else if (up_mode->valueint == 1)
    {
        printf("当前上报模式：变化上报\n");
        pthread_create(&tid, NULL, up_change, NULL);
    }
    else if (up_mode->valueint == 2)
    {
        printf("当前上报模式：周期上报\n");
        cJSON *period = cJSON_GetObjectItem(report, "period");
        PERIOD = period->valueint;
        pthread_create(&tid, NULL, up_period, NULL);
    }

    time_t t;
    char val[32];
    char stime[128];
    char sql[128];
    while (1)
    {
        sleep(3);
        struct tm *tm;
        time(&t);
        tm = localtime(&t);
        sprintf(stime, "%d-%02d-%02d %02d:%02d:%02d", tm->tm_year + 1900, tm->tm_mon + 1, tm->tm_mday, tm->tm_hour, tm->tm_min, tm->tm_sec);

        for (int i = 0; i < *total; i++)
        {
            if (shm_data[i].type == 1)
                sprintf(val, "%s", shm_data[i].new_val.b_val ? "true" : "false");
            else if (shm_data[i].type == 2)
                sprintf(val, "%d", shm_data[i].new_val.i_val);
            else
                sprintf(val, "%.2f", shm_data[i].new_val.f_val);
            sprintf(sql, "insert into history values(%d,'%s','%s');", shm_data[i].key, val, stime);
            if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != 0)
                printf("放入数据库失败\n");
        }
    }
    return 0;
}